package com.fast.bigdata.hourglass.controller;

import bigdata.hourglass.common.domain.Bean;
import bigdata.hourglass.common.utils.ZkUtils;
import com.alibaba.fastjson.JSONObject;
import com.fast.bigdata.hourglass.util.ActorUtil;
import com.fast.bigdata.hourglass.util.HdfsUtils;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import scala.collection.Seq;

import javax.servlet.http.HttpServletRequest;

/**
 * @author leo.jie (weixiao.me@aliyun.com)
 * @version 1.0
 * @organization CC
 * @website https://www.jlpyyf.com
 * @date 2019/12/8 10:22 下午
 * @since 1.0
 */
@RestController
@RequestMapping("/")
public class QueryController {
    @Autowired
    private ZkClient zkClient;
    @Autowired
    private ActorUtil actorUtils;
    @Autowired
    private Environment env;
    /**
     * 执行一个IQL
     */
    @PostMapping(value = "/query")
    public JSONObject execution(HttpServletRequest request, String iql, String mode, String variables, @RequestParam(defaultValue = "") String tag) {
        JSONObject resultObj = new JSONObject();
        resultObj.put("isSuccess", false);
        if ("".equals(iql.trim())) {
            resultObj.put("data", "LSql can't be empty...");
            return resultObj;
        }
        Seq<String> validEngines = ZkUtils.getValidChildren(zkClient, ZkUtils.validEnginePath(), tag);
        if (validEngines.size() == 0) {
            resultObj.put("data", "There is no available execution engine....");
            return resultObj;
        } else {
            String[] engineInfoAndActorName = validEngines.head().split("_");
            resultObj = actorUtils.queryActor(engineInfoAndActorName[0], engineInfoAndActorName[1],
                    new Bean.LSql(mode, iql, variables, null));
            return resultObj;
        }
    }

    /**
     * 获取结果
     */
    @PostMapping(value = "/getResult")
    public JSONObject getResult(HttpServletRequest request, String engineInfoAndGroupId) {
        JSONObject resultObj = new JSONObject();
        String validEngineByEngineInfo = getValidEngineByEngineInfo(engineInfoAndGroupId.split("_")[0]);
        if (validEngineByEngineInfo == null) {
            resultObj.put("isSuccess", false);
            resultObj.put("data", "当前未有可用的执行引擎...");
            return resultObj;
        } else {
            String[] engineInfoAndActorName = validEngineByEngineInfo.split("_");

            resultObj = actorUtils.queryActor(engineInfoAndActorName[0], engineInfoAndActorName[1],
                    new Bean.GetBatchResult(engineInfoAndGroupId));
            if (!resultObj.getString("status").equals("RUNNING")) {
                resultObj.put("success", resultObj.getBooleanValue("isSuccess"));
                if (resultObj.getString("dataType").equals("structuredData")) {
                    resultObj.put("data", HdfsUtils.readFileToString(resultObj.getString("hdfsPath"),
                            env.getProperty("hdfs.uri")));
                }
                return resultObj;
            } else {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return getResult(request, engineInfoAndGroupId);
            }
        }
    }

    /**
     * 获取指定节点上可用的actor
     *
     * @param engineInfo engine 信息
     * @return 可用的actor
     */
    private String getValidEngineByEngineInfo(String engineInfo) {
        Seq<String> validEngines = ZkUtils.getChildrenFilter(zkClient, ZkUtils.validEnginePath(), engineInfo);
        return validEngines.size() == 0 ? null : validEngines.head();
    }
}
